package com.gagakuai.concurrent;

import com.gagakuai.concurrent.locks.AbstactQueuedSychronier;
import com.gagakuai.concurrent.locks.Condition;
import com.gagakuai.concurrent.locks.ReentrantLock;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


/*
 * @program: common-starters
 * @description: 线程池执行器
 * <p>
 * <p>
 * 【ThreadPoolExecutor】通常被简称 为线程池，虽然与ForkJoinPool是兄弟关系，但两者的侧重点完全不同
 * <p>
 * ThreadPoolExecutor 由两部分组成  线程池【pool】和阻塞队列【BlockingQueue】
 * 线程池中存储着Worker[Workerpool] worker 兼具 任务和 线程的意义
 * 阻塞队列 中存储着 [work]
 * Worker可分为两类：【R】型Worker和【N】型Worker
 * 【R】型Worker携带了【任务】，它被执行完后，转为【N】型Worker。
 * 【N】型Worker自身不携带【任务】，它专门负责从阻塞队列中取出任务以便执行。
 * <p>
 * 【R】型线程：【R】型Worker中的线程域，一对一关系
 * 【N】型线程：【N】型Worker中的线程域，一对一关系
 * <p>
 * ★★★ 1. 在不产生歧义的情形下，下面可能会混用【Worker】与【线程】两词
 * ★★★ 2. 在前提1下，这里将：
 * 【R】型Worker或【R】型线程简称做【R】
 * 【N】型Worker或【N】型线程简称做【N】
 * <p>
 * 线程池状态： 状态机
 * -1 【运行】RUNNING   :   接收新线程，并可以处理阻塞任务
 * 0 【关闭】SHUTDOWN  : 不接收新线程，但可以处理阻塞任务，
 * 1 【停止】STOP      : 不接收新线程，不可以处理阻塞任务，且对正在执行的线程设置中断标记
 * 2 【完结】TIDYING   : 线程池空闲，阻塞队列为空，在调用terminated()后转入TERMINATED状态
 * 3 【终止】TERMINATED: terminated()方法已执行完
 * <p>
 * 状态转换：
 * shutdown()              线程池空闲，阻塞队列为空
 * ┌─────────────▶ SHUTDOWN ────────────┐
 * │                  │                 │            terminated()之后
 * RUNNING ──┤                  │ shutdownNow()   ├──▶ TIDYING ───────────────▶ TERMINATED
 * │                  ▼                 │
 * └─────────────▶   STOP   ────────────┘
 * shutdownNow()           线程池空闲
 * @author: houhong  // TODO: 2023/2/19  继续写线程池
 * @create: 2023-02-19 21:02
 */
public class ThreadPoolExecutor extends AbstractExecutorService {


    private static final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

    /*
     *  状态位
     *  由线程池状态标记的前3个比特位记录线程池的运行状态
     */
    /*
     * // 29
     */
    private static final int COUNT_BITS = Integer.SIZE - 3;
    /*
     * // 0b-0x 0001-FFF FFFF‬
     */
    private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
    /*
     * // 0b-0x 1110-000 0000，【运行】
     */
    private static final int RUNNING = -1 << COUNT_BITS;
    /*
     * 0b-0x 0000-000 0000，【关闭】
     */
    private static final int SHUTDOWN = 0 << COUNT_BITS;
    /*
     * 0b-0x 0010-000 0000，【停止】
     */
    private static final int STOP = 1 << COUNT_BITS;
    /*
     * 0b-0x 0100-000 0000，【完结】
     */
    private static final int TIDYING = 2 << COUNT_BITS;
    /*
     * 0b-0x 0110-000 0000，【终止】
     */
    private static final int TERMINATED = 3 << COUNT_BITS;

    /*
     * 启动超时设置
     */
    private static volatile boolean allowCoreThreadTimeOut;

    /*
     * 线程池锁,线程池状态标记改变的时候，使用
     */
    private final ReentrantLock mainLock = new ReentrantLock(Boolean.TRUE);
    /*
     * 线程池【条件队列】
     */
    private final Condition termination = mainLock.newCondition();


    /*
     * 线程池。存储核心线程池和非核心线程池
     */
    private final Set<Worker> workPool = new HashSet<>();

    /*
     * 时间单位
     */
    private TimeUnit unit;


    private ThreadFactory threadFactory;

    /*
     * 【拒绝策略】处理器
     */
    private volatile RejectedExecutionHandler handler;

    /*
     * 阻塞队列，队列周中存储
     */
    private final BlockingQueue<Runnable> workQueue;

    /*
     * 线程池的【核心阈值】 >= 0，一般用来限制【R】的数量
     */
    private volatile int corePoolSize;
    /*
     * 线程池的【最大阈值】 一般用来限制 【N】型的数量 （线程池满，阻塞队列也满时，限制【R】的数量）
     */
    private volatile int maxmumPoolSize;
    /*
     * 线程达到的最大数量
     */
    private int largestPoolSize;
    /*
     * 表示[N] 型线程的存活时间
     */
    private long keepAliveTime;

    /*
     * 默认的拒绝策略
     */
    public static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

    /*
     * 关闭线程池的权限
     */
    private static final RuntimePermission shutdownPerm = new RuntimePermission("modifyThread");

    /*==========构造器============*/
    public ThreadPoolExecutor(int corePoolSize,
                              int maxmumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workerBlockingQueue,
                              ThreadFactory threadFactory) {

        this(corePoolSize, maxmumPoolSize, keepAliveTime, unit, workerBlockingQueue, threadFactory, defaultHandler);
    }

    public ThreadPoolExecutor(int corePoolSize, int maxmumPoolSize, long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workerBlockingQueue,
                              RejectedExecutionHandler handler) {

        this(corePoolSize, maxmumPoolSize, keepAliveTime, unit, workerBlockingQueue, Executors.defaultThreadFactory(), handler);

    }


    public ThreadPoolExecutor(int corePoolSize,
                              int maxmumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit, BlockingQueue<Runnable> workerQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        /*
         * todo 值的判断
         */
        if (corePoolSize < 0
                || maxmumPoolSize < 0
                || corePoolSize < maxmumPoolSize
                || keepAliveTime < 0) {

            throw new IllegalArgumentException();
        }

        if (workerQueue == null
                || threadFactory == null || handler == null) {

            throw new IllegalArgumentException();
        }

        this.corePoolSize = corePoolSize;
        this.maxmumPoolSize = maxmumPoolSize;
        this.keepAliveTime = keepAliveTime;
        this.unit = unit;
        this.workQueue = workerQueue;
        this.threadFactory = threadFactory;
        this.handler = handler;

    }

    public ThreadFactory getThreadFactory() {
        return threadFactory;
    }


    private static int ctlOf(int runState, int workState) {

        /*
         *  位运算
         *  两个位都为0时，结果才为0
         * */
        return runState | workState;
    }


    /*
     * 执行给定的任务（视情形将其封装到线程池，或放入阻塞队列排队）
     */
    @Override
    public void execute(Runnable command) {

        //不执行空任务
        if (command == null) {
            throw new NullPointerException();
        }

        //线程池状态
        int state = ctl.get();

        //线程池的Worker数量在【核心阈值】以内
        if (workerCountof(state) < corePoolSize) {

            //添加一个worker 进入
            if (addWorker(command, true)) {
                // 执行成功，直接返回
                return;
            }

            state = ctl.get();


        }


        /*
         * 至此，有三种情形：
         * 1.线程池中的Worker数量超过了【核心阈值】（有些线程池将【核心阈值】设为0）
         * 2.【R】添加失败（由于线程争用，线程池状态发生了变化）
         * 3.【R】启动失败（少见）
         *
         * 此时，会尝试将任务添加到阻塞队列
         */

        // {-1} 如果线程池仍处于【运行】状态（可以接收新线程），且任务可以被成功添加到阻塞队列
        if (isRunning(state) && workQueue.offer(command)) {


            /*
             * 至此，任务成功进入了阻塞队列
             *
             * 为了执行阻塞队列中这个任务，
             * 于是，此时，尝试向线程池中添加一个【N】，
             * 目的是借助【N】间接执行阻塞队列中的任务
             *
             * 当然，这里如果没能成功添加【N】，也不要紧，
             * 因为稍后【R】执行完后，就会转变为【N】
             */

            int recheck = ctl.get();

            // {0123} 如果线程池已结束【运行】状态，这意味着线程池不再接收新线程了
            if (!isRunning(recheck)) {

                if (remove(command)) {
                    // 任务没能保存到阻塞队列，添加失败，执行【拒绝策略】
                    reject(command);
                }

            }else {

                /* {-1} 至此，说明线程池仍在【运行】状态 */


                // 如果线程池为空（通常出现在【核心阈值】为0的场景）
                if(workerCountof(state) == 0){
                    // 向线程池添加一个【N】，以便处理阻塞队列中的任务
                    addWorker(null, false);
                }

                /* 不为null的话不用管，因为自然有别的线程去处理阻塞队列的任务 */

            }

        }else {


            /*
             * 至此，有两种情形：
             * 1. {0123} 线程池已结束【运行】状态，这意味程池不再接收新的Worker了
             * 2. {-1}   线程池正处于【运行】状态，但任务没能成功进入阻塞队列（阻塞队列满了）
             *
             * 情形1：
             * 这种情形下，addWorker(command, false)往往返回false，
             * 即任务准备进入阻塞队列时，而且恰好线程池也已关闭，那么需要执行拒绝策略
             *
             * 情形2：
             * 这种情形下，可能是阻塞队列满了，此时改用【最大阈值】最为线程池的容量限制，
             * 重新尝试将任务包装为Worker放入了线程池，如果还是失败，则返回false
             * 比如【核心阈值】与【最大阈值】一样大时，这里往往是失败的
             */
             if(!addWorker(command,false)){
                 // 任务进入线程池失败，执行【拒绝策略】
                 reject(command);
             }
        }


    }


    /*
     * 一个Worker 兼具 [任务]和 【线程】的语义 它将待执行任务firsetWork封装成一个线程
     * 存储在自身的Thread域中
     * Worker可分为两类：【R】型Worker和【N】型Worker
     * 【R】型Worker携带了【任务】，它被执行完后，转为【N】型Worker（不断从阻塞队列中取出任务执行，没有任务时就阻塞）。
     * 【N】型Worker自身不携带【任务】，它专门负责从阻塞队列中取出任务以便执行。
     * <p>
     * 【R】型线程：【R】型Worker中的线程域，一对一关系
     * 【N】型线程：【N】型Worker中的线程域，一对一关系
     * <p>
     * 除此之外，Worker还兼具锁的功能，可在增减Worker以及执行Worker时加锁
     */
    private final class Worker extends AbstactQueuedSychronier implements Runnable {


        private static final long serialVersionUID = -1351338693575716769L;
        /*
         * 【工作线程】：后续执行firstTask【任务】的【线程】
         */
        final Thread workerThread;
        /*
         * 待 执行的任务
         */
        Runnable firstTask;
        /*
         * 记录当前Worker 执行的任务数
         */
        volatile long completedTasks;


        public Worker(Runnable firstTask) {
            // 每个任务的锁状态被初始化为-1，这使得工作线程在运行之前禁止中断
            setState(-1);

            this.firstTask = firstTask;
            this.workerThread = getThreadFactory()
                    .newThread(this);
        }

        //解锁，此处只是简单地将许可证数量置为0，以便后续可以对其加锁(会调用到tryRelease()方法)
        public void unlock() {

            release(1);
        }

        //加锁，此处只是简单地将许可证数量从0更新到1(会调用到tryAcquire()方法)
        public void lock() {
            //申请独占锁
            acquire(1);
        }

        @Override
        public void run() {
            runWorker(this);
        }

        /*
         * @param worker
         * @return
         * @Author houhong
         * @Description //TODO
         * <p>
         * 由[工作线程]调用，执行线程池中的任务，
         * <p>
         * 如果当前Worker是【R】型，则取出【R】型Worker中的任务执行，之后，【R】型Worker转变为【N】型Worker；
         * 如果当前Worker(已经)是【N】型Worker，则需要不断从线程池的阻塞队列中取出待执行任务。
         * <p>
         * 如果未启用超时设置，则没有待执行任务时，该工作线程会被阻塞；
         * 如果超时设置已被激活，那么没有待执行任务时，该工作线程会退出。
         * @Date 10:20 下午 2023/2/22
         */
        final void runWorker(Worker worker) {

            Thread currentThread = Thread.currentThread();

            //获取worker中的任务
            Runnable runnableTask = worker.firstTask;
            //置空 (如果该Worker是【R】型Worker，此处相当于将其从【R】型转变为【N】型)
            worker.firstTask = null;

            /*
             *  解锁，此处只是简单地将许可证数量置为0
             *  每个Worker的锁状态被初始化为-1，
             *  这里要先将锁状态重置为0，以便可以对Worker正常加锁
             *
             */
            worker.unlock();
            //是否异常结束
            boolean completedAbruptly = true;

            try {
                /*
                 *  分为两种情况，
                 *      1：当第一次进来的时候，此时是N型任务
                 *      2：第二次进来的时候，为n型任务
                 *
                 */
                while (true) {

                    if (runnableTask == null) {

                        //从queue中获取任务
                        runnableTask = getTask();

                        //阻塞队列中也没有任务了
                        if (runnableTask == null) {
                            break;
                        }
                    }
                    //加锁，这里会阻塞在CHL队列中，直到申请到锁
                    worker.lock();

                    /*
                     *     确保：
                     *      1 线程池在(-1,0) 状态时，线程未被中断
                     *      2：线程池在{1,2,3}状态时，线程被中断
                     *
                     */
                    if (runStateAtLeast(ctl.get(), STOP)) {
                        //当前线程不在中断中，需要为当前线程设置中断
                        if (!Thread.interrupted()) {
                            currentThread.interrupt();
                        }

                    } else {
                        // 线程池在(-1,0) 状态时 -- 表示运行中，关闭 的状态，在这两种状态中，必须确保处理完任务。所以要清除中断标记
                        // （静态）测试当前线程是否已经中断，线程的中断状态会被清除
                        Thread.interrupted();
                    }

                    try {

                        // 钩子函数。在任务执行之前处理，然后如果要拓展的话，只需要基础 ThreadPoolExecuteService
                        //然后重写该方法
                        beforeExecute(currentThread, runnableTask);

                        runnableTask.run();
                        // 后置钩子
                        afterExecute(currentThread, null);


                    } catch (Throwable ex) {

                        afterExecute(currentThread, ex);
                        throw ex;

                    } finally {
                        //置空 任务，以方便重新运行
                        runnableTask = null;
                        //当前完成任务++
                        worker.completedTasks++;
                        //解锁
                        worker.unlock();
                    }


                }


            } finally {

                //// TODO: 2023/2/23
                processWorkerExit(worker, completedAbruptly);
            }


        }

        /*
         * 线程池中的某个Worker结束，除了需要将该Worker从线程池移除，还要保证阻塞队列中的阻塞任务后续有人处理
         *
         */
        private void processWorkerExit(Worker worker, boolean completedAbruptly) {

            //如果工作线程是非正常退出的
            if (completedAbruptly) {

                // 原子地递【减】线程池中工作线程(Worker)的数量，并更新线程池状态标记
                decrementWorkerCount();
            }


            mainLock.lock();
            // 记录线程池累计执行的任务数量
            try {
                completedTasks += worker.completedTasks;
                // 从线程池中移除该Worker
                workPool.remove(worker);
            } finally {
                mainLock.unlock();
            }


            //todo 实现未写完成
            tryTerminate();


            int state = ctl.get();


            /*
             *  至此说明工作线程被异常结束，或者，工作线程数量已经低于最小预留值的保障；
             *  此时需要专门添加一个【N】型Worker到线程池，以便处理阻塞队列中的任务
             */


        }


        /*
         * @param
         * @return {@link Runnable}
         * @Author houhong
         * @Description //TODO 从阻塞queue中 获取任务
         * <p>
         * N】型Worker通过此方法，从阻塞队列中取出任务，如果没取到，则返回null
         * <p>
         * 当返回值为null,往往伴随着线程池中的worker的递减
         * <p>
         * 如果未启用超时设置，那么该方法会一直阻塞，直到从阻塞队列中取到任务
         * 如果启用了超时设置，则必须在指定时间内取到任务，否则返回null
         * @Date 10:46 下午 2023/2/22
         */
        public Runnable getTask() {

            boolean timeout = false;
            for (; ; ) {

                int state = ctl.get();

                //{0,1,2,3}状态，表示 【线程池】 不再接收新的线程了
                if (runStateAtLeast(state, SHUTDOWN)) {

                    //当线程池状态为 {1,2,3} 表示线程池不能处理阻塞队列中的阻塞任务了，或者
                    //可以处理,但是线程池为空
                    if (runStateAtLeast(state, STOP) || workQueue.isEmpty()) {

                        //原子性的更改线程池中的worker数量
                        decrementWorkerCount();

                        //返回null 意味着 该N型 worker 随后结束
                        return null;

                    }

                }


                /*到现在，说明线程池有能力处理阻塞任务，并且线程池不为空*/


                //获取线程池中的Worker数量
                int workerCount = workerCountof(state);

                /*
                 *  是否启用超时设置
                 *  1： 主动设置 可以通过设置allowCoreThreadTimeOut==true来实现
                 *  2：自动设置，发生在线程池中的Worker数量超过【核心阈值】时
                 *
                 * */
                boolean timed = allowCoreThreadTimeOut || workerCount > corePoolSize;

                /*
                 *  当worker数量大于最大阈值（线程争用严重时会有此种情形）
                 *
                 */
                if (workerCount > maxmumPoolSize
                        // 或者启用了超时设置，且已经超时
                        || (timed || timeout)) {

                    // 如果线程池中至少有一个Worker
                    if (workerCount > 1
                            // 或者阻塞队列为空
                            || workQueue.isEmpty()) {

                        /*
                         * 至此，归纳为两种情形：
                         * 1. 线程池中的Worker数量超过【最大阈值】
                         * 2. 线程池启用了超时设置，并已经超时
                         */


                        /*
                         * 原子地递【减】线程池中工作线程(Worker)的数量，并更新线程池状态标记
                         *
                         * 递减的原因是Worker数量过多了...
                         */
                        if (compareAndDecrementWorkerCount(state)) {

                            //返回null 意味着 该N型 worker 随后结束
                            return null;
                        }
                        continue;
                    }

                }


                /*==== 到这里删除了多余的【N】型 worker*/
                try {
                    Runnable task;

                    //启动超时
                    if (timed) {
                        // 出队，从阻塞队列取出任务，不满足出队条件时阻塞一段时间，如果在指定的时间内没有成功拿到元素，则返回null
                        task = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
                    } else {
                        task = workQueue.take();
                    }


                    if (task != null) {
                        return task;
                    }
                    // 至此，说启用了超时设置，且确实已经超时了（【N】过时了）
                    timeout = true;
                } catch (InterruptedException retry) {
                    timeout = true;
                } finally {
                }

            }


        }


    }

    public boolean addWorker(Runnable task, boolean core) {


        for (int state = ctl.get(); ; ) {

        }
    }

    /*
     * @param thread       线程
     * @param runnableTask 任务
     * @return
     * @Author houhong
     * @Description //TODO  做的一个钩子函数
     * @Date 11:10 下午 2023/2/22
     */
    protected void beforeExecute(Thread thread, Runnable runnableTask) {


    }

    /*
     * 做一个后置的钩子函数
     */
    protected void afterExecute(Thread thread, Throwable ex) {


    }

    public static boolean compareAndDecrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect - 1);
    }

    // 从阻塞队列中移除之前添加的任务
    public boolean remove(Runnable task) {
        // 移除任务
        boolean removed = workQueue.remove(task);

        // 尝试让线程池进入【终止】状态
        tryTerminate(); // In case SHUTDOWN and now empty

        return removed;
    }

    /*
     * 尝试让线程池进入{3}【终止】状态
     */
    final void tryTerminate() {


        for (; ; ) {


        }


    }

    /*
     * 为指定的动作（任务）执行【拒绝策略】
     *
     * 常见触发时机：
     * 1.任务【准备】进入阻塞队列时，恰好发现线程池已关闭
     * 2.任务进入阻塞队列【失败】（线程池仍在运行）
     * 3.任务【已经】进入阻塞队列，但后续发现线程池关闭了，
     *   此时不仅要执行拒绝策略，还要将该任务从阻塞队列中移除
     */
    final void reject(Runnable command) {
        handler.rejectExecution(command, this);
    }

    /*
     * @return
     * @Author houhong
     * @Description //TODO  原子性的递减线程池中的worker
     * @Date 8:59 下午 2023/2/25
     */
    public void decrementWorkerCount() {

        ctl.addAndGet(-1);
    }

    public static boolean runStateAtLeast(int workPoolState, int flag) {

        return workPoolState >= flag;
    }

    private static boolean runStateLessThan(int workerPoolState, int flag) {
        return workerPoolState < flag;
    }

    private static boolean isRunning(int workerPoolState) {
        return workerPoolState < SHUTDOWN;
    }

    /*
     *  返回线程池中的工作线程(Worker)数量（位于线程池状态标记的后29位）
     */
    public static int workerCountof(int workerPoolState) {
        return workerPoolState & COUNT_MASK;
    }


    /*=====================预定义的拒绝策略==================*/

    /*
     * @Author houhong
     * @Description //TODO 丢弃任务，抛出异常
     * @Date 11:47 下午 2023/2/22
     * @return {@link null}
     */
    public static class AbortPolicy implements RejectedExecutionHandler {


        public AbortPolicy() {


        }

        @Override
        public void rejectExecution(Runnable r, ThreadPoolExecutor executor) {

            throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + executor.toString());
        }
    }


}